package ar.edu.itba.pod.Legajo47342.market;

import java.rmi.RemoteException;

import ar.edu.itba.pod.Legajo47342.Node;
import ar.edu.itba.pod.Legajo47342.communication.payload.ResourceRequestPayloadImpl;
import ar.edu.itba.pod.simul.communication.Message;
import ar.edu.itba.pod.simul.communication.MessageType;
import ar.edu.itba.pod.simul.communication.payload.Payload;
import ar.edu.itba.pod.simul.local.LocalMarket;
import ar.edu.itba.pod.simul.market.Resource;
import ar.edu.itba.pod.simul.market.ResourceStock;

import com.google.common.collect.ConcurrentHashMultiset;
import com.google.common.collect.Multiset;

public class MarketImpl extends LocalMarket {

	private Multiset<Resource> remoteSelling;
	private Node node;
	private Double transPerSec;
	
	public MarketImpl(Node node) {
		super();
		this.node = node;
		this.remoteSelling = ConcurrentHashMultiset.create();
		this.transPerSec = (double) 0;
	}


	@Override
	protected void matchBothEnds() {
		int transfered;
		
		for (ResourceStock buyer : buying.elementSet()) {
			//System.out.println("buyer name: "+buyer.name());
			for (ResourceStock seller : selling) {
				if (buyer.resource().equals(seller.resource())) {
					transfered = transfer(buyer, seller);
					if (transfered > 0) {
						System.out.println(buyer.name()+" bought "+
								transfered+" "+buyer.resource().name()+" from "+seller.name());
					}
				}
			}
			
			if (buying.count(buyer) > 0) {
				for (Resource r :remoteSelling.elementSet()) {
					if (buyer.resource().equals(r)) {
						transfered = transferRemote(buyer, r);
						if (transfered > 0) {
							System.out.println(buyer.name()+" bought "+
									transfered+" "+buyer.resource().name());
						}
					}
				}
			}
			
			if (buying.count(buyer) > 0) {
				//todavia necesita recursos, los pido a la red
				Payload p = new ResourceRequestPayloadImpl(buyer.resource(), buying.count(buyer));
				Message m = new Message(node.getNodeId(), System.currentTimeMillis(), MessageType.RESOURCE_REQUEST, p);
				try {
					node.getManager().getGroupCommunication().broadcast(m);
				} catch (RemoteException e) {
					e.printStackTrace();
				}
			}
		}
	}
	
	protected int transferRemote(ResourceStock buyer, Resource seller) {
		while(true) {
			int wanted = buying.count(buyer);
			int available = remoteSelling.count(seller);
			int transfer = Math.min(available, wanted);
			System.out.println("TRANSFER: "+transfer);
			
			if (transfer == 0) {
				return 0;
			}
	
			boolean procured = remoteSelling.setCount(seller, available, available - transfer);
			if (procured) {
				boolean sent = buying.setCount(buyer, wanted, wanted - transfer);
				if (sent) {
					try {
						buyer.add(transfer);
					}
					catch (RuntimeException e) {
						// market takes care of what was sold. 
						// To fully solve this case, 2PCommit or 3PC is required. Is it worth?   
						buying.remove(buyer, transfer);
						continue;
					}
					//logTransfer(seller, buyer, transfer);
					transactionCount++;
					
					return transfer;
				}
				else {
					// Compensation. restore what we took from the order!
					remoteSelling.add(seller, transfer);
				}
			}
			// Reaching here mean we hit a race condition. Try again.
		}
	}
	
	public Multiset<ResourceStock> getSelling(){
		return selling;
	}
	
	public Multiset<Resource> getRemoteSelling(){
		return remoteSelling;
	}


	public void addData(double transactionsPerSecond) {
		transPerSec += transactionsPerSecond;
		
	}

}
